package com.sdy.mq.service.impl;

import com.sdy.mq.config.RocketMqConfig;
import com.sdy.mq.service.MqProducerService;
import com.sdy.mq.service.MqSender;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.UUID;

@Slf4j
@Service
public class MqProducerServiceImpl implements MqProducerService {
    @Autowired
    private RocketMqConfig rocketMqConfig;
    @Autowired
    private DefaultMqSender defaultMqSender;
    @Autowired
    private OrderedMqSender orderedMqSender;
    
    private MqSender getSender(Object ...args) {
        if (args.length > 0 && (Boolean) args[0]) {
            return orderedMqSender;
        } else {
            return defaultMqSender;
        }
    }
    
    @Override
    public SendResult send(String tag, String content, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(rocketMqConfig.getTopic(), tag, key, content.getBytes(RemotingHelper.DEFAULT_CHARSET));
        log.debug("Produce message: data=[{}]", byteToString(msg));
        return getSender(args).send(msg);
    }

    @Override
    public SendResult send(String tag, byte[] content, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(rocketMqConfig.getTopic(), tag, key, content);
        log.debug("Produce message: data=[{}]", byteToString(msg));
        return getSender(args).send(msg);
    }

    @Override
    public SendResult send(String tag, String content, Integer delayLevel, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(rocketMqConfig.getTopic(), tag, key, content.getBytes(RemotingHelper.DEFAULT_CHARSET));
        msg.setDelayTimeLevel(delayLevel);
        log.debug("Produce message: data=[{}]", byteToString(msg));
        return getSender(args).send(msg);
    }

    @Override
    public SendResult send(String tag, byte[] content, Integer delayLevel, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(rocketMqConfig.getTopic(), tag, key, content);
        msg.setDelayTimeLevel(delayLevel);
        log.debug("Produce message: data=[{}]", byteToString(msg));
        return getSender(args).send(msg);
    }

    @Override
    public SendResult sendAsync(String tag, String content, SendCallback sendCallback, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(rocketMqConfig.getTopic(), tag, key, content.getBytes(RemotingHelper.DEFAULT_CHARSET));
        log.debug("Produce message: data=[{}]", byteToString(msg));
        getSender(args).send(msg, sendCallback);
        return null;
    }

    @Override
    public SendResult sendAsync(String tag, byte[] content, SendCallback sendCallback, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(rocketMqConfig.getTopic(), tag, key, content);
        log.debug("Produce message: data=[{}]", byteToString(msg));
        getSender(args).send(msg, sendCallback);
        return null;
    }

    @Override
    public SendResult tSend(String topic, String tag, String content, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(topic, tag, key, content.getBytes(RemotingHelper.DEFAULT_CHARSET));
        log.debug("Produce message: data=[{}]", byteToString(msg));
        return getSender(args).send(msg);
    }

    @Override
    public SendResult tSend(String topic, String tag, byte[] content, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(topic, tag, key, content);
        log.debug("Produce message: data=[{}]", byteToString(msg));
        return getSender(args).send(msg);
    }

    @Override
    public SendResult tSend(String topic, String tag, String content, Integer delayLevel, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(topic, tag, key, content.getBytes(RemotingHelper.DEFAULT_CHARSET));
        msg.setDelayTimeLevel(delayLevel);
        log.debug("Produce message: data=[{}]", byteToString(msg));
        return getSender(args).send(msg);
    }

    @Override
    public SendResult tSend(String topic, String tag, byte[] content, Integer delayLevel, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(topic, tag, key, content);
        msg.setDelayTimeLevel(delayLevel);
        log.debug("Produce message: data=[{}]", byteToString(msg));
        return getSender(args).send(msg);
    }

    @Override
    public SendResult tSendAsync(String topic, String tag, String content, SendCallback sendCallback, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(topic, tag, key, content.getBytes(RemotingHelper.DEFAULT_CHARSET));
        log.debug("Produce message: data=[{}]", byteToString(msg));
        getSender(args).send(msg, sendCallback);
        return null;
    }

    @Override
    public SendResult tSendAsync(String topic, String tag, byte[] content, SendCallback sendCallback, Object ...args) throws Exception {
        String key = generateKey();
        Message msg = new Message(topic, tag, key, content);
        log.debug("Produce message: data=[{}]", byteToString(msg));
        getSender(args).send(msg, sendCallback);
        return null;
    }

    /**
     * 生成消息key
     */
    private String generateKey() {
        return UUID.randomUUID().toString().replaceAll("-", "");
    }

    /**
     * Message打印
     */
    private String byteToString(Message message) {
        return "Message{" +
                "topic='" + message.getTopic() +
                ", tags='" + message.getTags() +
                ", key='" + message.getKeys() +
                ", flag=" + message.getFlag() +
                ", properties=" + message.getProperties() +
                ", body=[byte[...]]" +
                ", transactionId='" + message.getTransactionId() +
                '}';
    }
}
